package com.thread.pool.core.common.executor;

import com.thread.pool.core.common.constant.ThreadPoolConstants;
import com.thread.pool.domain.param.ThreadPoolParam;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.core.env.MutablePropertySources;
import org.springframework.core.env.PropertySource;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 线程池服务类
 * 可动态修改线程池的数量
 */
@Component
public class ThreadPoolManager {


    private static final Logger log = LoggerFactory.getLogger(ThreadPoolManager.class);

    /**
     * 说明：下面这些常量我是根据AsyncTask的源码配置的，大家可以根据自己需求自行配置
     * 根据cpu的数量动态的配置核心线程数和最大线程数
     */
    private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
    /**
     * 默认核心线程数 = CPU核心数 * 2
     */
    private static final int DEFAULT_CORE_POOL_SIZE = CPU_COUNT * 2;
    /**
     * 默认线程池最大线程数 = CPU核心数 * 4
     */
    private static final int DEFAULT_MAXIMUM_POOL_SIZE = CPU_COUNT * 4 + 1;


    /**
     * 默认队列长度 1000
     */
    private static final int DEFAULT_QUEUE_SIZE = 1000;

    /**
     * 默认非核心线程闲置时超时1s
     */
    private static final int DEFAULT_KEEP_ALIVE = 1;

    /**
     * 线程计数器
     */
    private static final AtomicInteger THREAD_COUNTER = new AtomicInteger(1);
    /**
     * 线程池标识前缀
     */
    private static final String POOL_PREFIX = "service-thread-pool";

    /**
     * 核心线程数，如果配置文件存在使用该核心线程数
     */
    @Value("${thread.pool.core.size:}")
    private Integer coreSize;

    /**
     * 最大线程数,该配置优先
     */
    @Value("${thread.pool.max.size:}")
    private Integer maxSize;

    /**
     * 队列数，该配置优先
     */
    @Value("${thread.pool.queue.size:}")
    private Integer queueSize;

    /**
     * 非核心线程闲置时超时 单位为秒
     */
    @Value("${thread.pool.keep.alive:}")
    private Integer keepAlive;

    /**
     * 线程服务实例，可以用于线程间传递临时变量
     */
    private ExecutorService executorService;

    /**
     * 当前核心线程数
     */
    private Integer currentCoreSize;

    /**
     * 当前最大线程数
     */
    private Integer currentMaxPoolSize;

    /**
     * 非核心线程闲置时超
     */
    private Integer currentKeepAlive;

    /**
     * 非核心线程闲置时超
     */
    private Integer currentQueueSize;

    /**
     * 初始化线程池服务类
     */
    private void initExecutorService() {
        final int poolFlag = THREAD_COUNTER.getAndIncrement();
        /**
         * corePoolSize:核心线程数
         * maximumPoolSize：线程池所容纳最大线程数(workQueue队列满了之后才开启)
         * keepAliveTime：非核心线程闲置时间超时时长
         * unit：keepAliveTime的单位
         * workQueue：等待队列，存储还未执行的任务
         * threadFactory：线程创建的工厂
         * handler：异常处理机制
         */
        currentCoreSize = coreSize == null ? DEFAULT_CORE_POOL_SIZE : coreSize;
        currentMaxPoolSize = maxSize == null ? DEFAULT_MAXIMUM_POOL_SIZE : maxSize;
        currentKeepAlive = keepAlive == null ? DEFAULT_KEEP_ALIVE : keepAlive;
        currentQueueSize = queueSize == null ? DEFAULT_QUEUE_SIZE : queueSize;

        ThreadPoolExecutor executor = new ThreadPoolExecutor(currentCoreSize, currentMaxPoolSize,
                currentKeepAlive, TimeUnit.SECONDS, new ResizableCapacityLinkedBlockIngQueue<>(currentQueueSize),
                getThreadFactory(poolFlag),
                getCallerRunsPolicy()) {
            // 重写完成之后的方法
            @Override
            protected void afterExecute(Runnable r, Throwable t) {
                super.afterExecute(r, t);
                if (t != null) {
                    log.error("service-thread-pool# {} throw exception when execute task ", POOL_PREFIX + "@" + poolFlag, t);
                }
            }
        };
        executorService = executor;
    }

    /**
     * 获取创建线程的工厂
     * @param poolFlag
     * @return
     */
    private ThreadFactory getThreadFactory(int poolFlag){
        return  new ThreadFactory() {
            private AtomicInteger innerCounter = new AtomicInteger(1);

            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName(POOL_PREFIX + "@" + poolFlag + "-" + innerCounter.getAndIncrement());
                thread.setPriority(Thread.NORM_PRIORITY);
                thread.setDaemon(false);
                return thread;
            }
        };
    }



    /**
     * 获取拒绝策略
     * @return
     */
    private ThreadPoolExecutor.CallerRunsPolicy getCallerRunsPolicy(){
        return new ThreadPoolExecutor.CallerRunsPolicy();
    }


    /**
     * 使用线程池，线程池中线程的创建完全是由线程池自己来维护的，我们不需要创建任何的线程
     * 我们所需要做的事情就是往这个池子里面丢一个又一个的任务
     *
     * @param r 执行线程  可以传入实现Runnable的Future
     */
    public void execute(Runnable r) {
        if (executorService == null) {
            initExecutorService();
        }
        // 把一个任务丢到了线程池中
        executorService.execute(r);
    }



    /**
     * 提交数据到线程池
     * @param callable
     */
    public void submit(Callable callable) {
        if (executorService == null) {
            initExecutorService();
        }
        // 把一个任务提交到线程池中
        executorService.submit(callable);
    }



    /**
     * 执行并去获取结果
     * 泛型建议使用 Result<T> 对象
     * 同时异常的处理尽量在调用的方法中进行处理，不要在future.get()的时候有异常
     * @param callableList 自定义的callableList接口
     * @param <T> 泛型
     * @return 执行结果
     */
    public <T> List<T> getExecuteResult(List<Callable<T>> callableList)  {
        if (executorService == null) {
            initExecutorService();
        }

        List<Future<T>> futures = new ArrayList<>(callableList.size());
        List<T> result = new ArrayList<>(callableList.size());
        callableList.forEach(callable -> {
            futures.add(executorService.submit(callable));
        });
        for (Future<T> future : futures) {
            try {
                T t = future.get();
                result.add(t);
            } catch (InterruptedException e) {
                log.error("线程执行中断异常",e);
                throw new RuntimeException("线程执行中断异常");
            } catch (ExecutionException e) {
                log.error("线程执行异常",e);
                throw new RuntimeException("线程执行异常");
            }
        }
        return result;
    }

    /**
     * 重新处理线程池大小
     * @param context
     */
    public void reset(ConfigurableApplicationContext context){
        log.info("收到配置文件变更消息，重新设置线程池的大小>>>");
        MutablePropertySources propertySources = context.getEnvironment().getPropertySources();

        Integer coreSizeParam = null;
        Integer maxPoolSizeParam = null;
        Integer queueSizeParam = null;
        for (PropertySource<?> propertySource : propertySources) {
            if(propertySource.containsProperty(ThreadPoolConstants.THREAD_POOL_CORE_SIZE)){
                coreSizeParam = Integer.parseInt(propertySource.getProperty(ThreadPoolConstants.THREAD_POOL_CORE_SIZE).toString());
            }
            if(propertySource.containsProperty(ThreadPoolConstants.THREAD_POOL_MAX_SIZE)){
                maxPoolSizeParam = Integer.parseInt(propertySource.getProperty(ThreadPoolConstants.THREAD_POOL_MAX_SIZE).toString());
            }
            if(propertySource.containsProperty(ThreadPoolConstants.THREAD_POOL_QUEUE_SIZE)){
                queueSizeParam = Integer.parseInt(propertySource.getProperty(ThreadPoolConstants.THREAD_POOL_QUEUE_SIZE).toString());
            }
        }
        ThreadPoolParam threadPoolParam = ThreadPoolParam.build()
                .setCoreSize(coreSizeParam)
                .setMaxSize(maxPoolSizeParam)
                .setQueueSize(queueSizeParam);
        reset(threadPoolParam);
    }


    /**
     * 重新处理线程池大小
     */
    public void reset(ThreadPoolParam dto){
        log.info("重新设置线程池的大小,coreSize={},maxSize={},queueSize={}");
        currentCoreSize = currentCoreSize == null ? DEFAULT_CORE_POOL_SIZE : currentCoreSize;
        currentMaxPoolSize = currentMaxPoolSize == null ? DEFAULT_MAXIMUM_POOL_SIZE : currentMaxPoolSize;
        currentQueueSize = currentQueueSize == null ? DEFAULT_QUEUE_SIZE : currentQueueSize;

        Integer coreSizeParam = dto.getCoreSize();
        Integer maxPoolSizeParam = dto.getMaxSize();
        Integer queueSizeParam = dto.getQueueSize();
        ThreadPoolExecutor excutor = (ThreadPoolExecutor) this.executorService;
        // 设置核心线程数
        if(coreSizeParam != null && coreSizeParam > currentCoreSize){
            excutor.setCorePoolSize(coreSizeParam);
        }
        // 设置最大线程数
        if(maxPoolSizeParam != null && maxPoolSizeParam > currentMaxPoolSize){
            excutor.setMaximumPoolSize(maxPoolSizeParam);
        }
        // 设置等待队列的大小
        if(queueSizeParam != null && queueSizeParam > currentQueueSize){
            BlockingQueue<Runnable> queue = excutor.getQueue();
            ((ResizableCapacityLinkedBlockIngQueue)(queue)).setCapacity(queueSizeParam);
        }
        excutor.prestartAllCoreThreads();
    }


    public ExecutorService getExecutorService(){
        return executorService;
    }
}
